package com.kingsoft.dc.khaos.module.spark.util

import java.util.{ArrayList, List}

import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.extender.meta.api.{AuthRequest, DmRequest}
import com.kingsoft.dc.khaos.extender.meta.model.MetaParams
import com.kingsoft.dc.khaos.extender.meta.model.auth.DmAuth
import com.kingsoft.dc.khaos.extender.meta.model.col.DmTableColumn
import com.kingsoft.dc.khaos.extender.meta.model.ds._
import com.kingsoft.dc.khaos.extender.meta.model.table.DmTable
import com.kingsoft.dc.khaos.extender.meta.utils.DmMetaUtils
import com.kingsoft.dc.khaos.module.spark.constants.TableType.TableType
import com.kingsoft.dc.khaos.module.spark.constants.{SchedulerConstants, TableType}
import com.kingsoft.dc.khaos.module.spark.model.MetaDataEntity
import com.kingsoft.dc.khaos.module.spark.sink.SinkStrategy
import com.kingsoft.dc.khaos.module.spark.source.SourceStrategy
import com.kingsoft.dc.khaos.util.Logging

/**
  * Created by jing on 19/7/2.
  */
object NewMetaUtils extends Logging {

  def buildAuthRequest(kc: KhaosContext): DmRequest = {
    val request = new DmRequest()
    request.setUrl(kc.conf.getString(SchedulerConstants.META_RIGHT_QUERY_URL_KEY))
    val authRequest: AuthRequest = new AuthRequest()
    authRequest.setAuthUrl(kc.conf.getString(SchedulerConstants.AUTH_SERVER_KEY))
    authRequest.setAuthProjectId(kc.conf.getInt(SchedulerConstants.PROJECT_ID))
    request.setAuthRequest(authRequest)
    request
  }

  def buildMetaRequest(kc: KhaosContext): DmRequest = {
    val request = new DmRequest()
    request.setUrl(kc.conf.getString(SchedulerConstants.META_QUERY_URL_KEY))
    val authRequest: AuthRequest = new AuthRequest()
    authRequest.setAuthUrl(kc.conf.getString(SchedulerConstants.AUTH_SERVER_KEY))
    authRequest.setAuthProjectId(kc.conf.getInt(SchedulerConstants.PROJECT_ID))
    request.setAuthRequest(authRequest)
    request
  }

  def checkReadAuth(kc: KhaosContext, dbName: String, tblName: String, className: String, authParam: String): Boolean = {
    val pivlist = new ArrayList[String]()
    pivlist.add("select")
    val result = checkAuth(kc, dbName, tblName, className, authParam, pivlist)
    result
  }

  def checkWriteAuth(kc: KhaosContext, dbName: String, tblName: String, className: String, authParam: String): Boolean = {
    val pivlist = new ArrayList[String]()
    pivlist.add("select")
    pivlist.add("insert")
    val result = checkAuth(kc, dbName, tblName, className, authParam, pivlist)
    result
  }

  /**
    * 统一鉴权方法
    * @param kc
    * @param dbName    //库名
    * @param tblName   //表名
    * @param className  //鉴权实现类名
    * @param authParam //鉴权json
    * @param pivlist  //权限列表,比如List("insert","select")
    * @return
    */
  def checkAuth(kc: KhaosContext, dbName: String, tblName: String, className: String, authParam: String, pivlist: ArrayList[String]): Boolean = {
    val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)
    val projectId = kc.conf.getString(SchedulerConstants.PROJECT_ID, "0")
    val projectName = kc.conf.getString(SchedulerConstants.PROJECT_NAME, null)
    //logInfo(s"checkAuth authParam ==> $authParam")
    val request = buildAuthRequest(kc)
    val metaParams = new MetaParams(authParam)
    metaParams.buildAuthParams(runEnv, projectId, projectName, dbName, tblName, pivlist);
    request.setMetaParams(metaParams)
    val auth: DmAuth = DmMetaUtils.checkProjectAuth(className, request, classOf[DmAuth])
    val checkResutlt = auth.getTotalPrivs
    if (!checkResutlt) {
      throw new Exception("权限校验未通过")
    }
    checkResutlt
  }

  /**
    * 获取所有元数据信息，连接信息，数据表属性，数据表列属性
    * @param kc
    * @param dbName
    * @param tblName
    * @param className
    * @param metaParamsJson
    * @param classType
    * @tparam T
    * @return
    */
  def getMeta[T](kc: KhaosContext, dbName: String, tblName: String, className: String, metaParamsJson: String, classType: T, tblType: TableType): MetaDataEntity = {
    val connect = getDsConnect(kc, dbName, tblName, className, metaParamsJson, classType)
    val table = getTableProps(kc, dbName, tblName, className, metaParamsJson, tblType);
    val columns = getTableColumnProps(kc, dbName, tblName, className, metaParamsJson, tblType);
    val mde = new MetaDataEntity
    mde.setConnect(connect)
    mde.setTableEntiy(table)
    mde.setColumnEntiy(columns)
    mde
  }

  /**
    * 获取数据源连接信息
    * @param kc
    * @param dbName
    * @param tblName
    * @param className
    * @param metaParamsJson
    * @param classType
    * @tparam T
    * @return
    */
  def getDsConnect[T](kc: KhaosContext, dbName: String, tblName: String, className: String, metaParamsJson: String, classType: T): Connect = {
    val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)
    val jobProjectId = kc.conf.getInt(SchedulerConstants.PROJECT_ID, 0)
    val jobProjectName = kc.conf.getString(SchedulerConstants.PROJECT_NAME, null)

    //获取物理地址
    var metaRequest = buildMetaRequest(kc)
    var metaParams = new MetaParams(metaParamsJson)
    metaParams.setJobProjectId(jobProjectId)
    //metaParams.setProjectName(jobProjectName)
    if (classType.isInstanceOf[SourceStrategy]) {
      metaParams.buildReadDatasourceParams(runEnv, dbName, tblName)
    } else if (classType.isInstanceOf[SinkStrategy]) {
      metaParams.buildWriteDatasourceParams(runEnv, dbName, tblName)
    }
    metaRequest.setMetaParams(metaParams)

    val ds = DmMetaUtils.getDatasource(className, metaRequest, classOf[DmDatasource])
    ds.getConnect
  }
  /**
    * 获取数据表属性
    * @param kc
    * @param dbName
    * @param tblName
    * @param className
    * @param metaParamsJson
    * @tparam T
    * @return
    */
  def getTableProps[T](kc: KhaosContext, dbName: String, tblName: String, className: String, metaParamsJson: String, tableType: TableType): DmTable = {
    val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)
    val jobProjectId = kc.conf.getInt(SchedulerConstants.PROJECT_ID, 0)
    val jobProjectName = kc.conf.getString(SchedulerConstants.PROJECT_NAME, null)

    //获取物理地址
    var metaRequest = buildMetaRequest(kc)
    var metaParams = new MetaParams(metaParamsJson)
    metaParams.setJobProjectId(jobProjectId);
    metaParams.setJobProjectName(jobProjectName);
    metaRequest.setMetaParams(metaParams)
    if (tableType.equals(TableType.NOSQL)) {
      metaParams.buildNosqlTableParams(runEnv, dbName, tblName)
    } else if (tableType.equals(TableType.RELATION)) {
      metaParams.buildTableParams(runEnv, dbName, tblName)
    } else if (tableType.equals(TableType.FILE)) {
      metaParams.buildFileTableParams(runEnv, dbName, tblName)
    } else {
      metaParams.buildNosqlTableParams(runEnv, dbName, tblName)
    }

    val table = DmMetaUtils.getTable(className, metaRequest, classOf[DmTable])
    table
  }

  /**
    * 获取数据表 列属性
    * @param kc
    * @param dbName
    * @param tblName
    * @param className
    * @param metaParamsJson
    * @return
    */
  def getTableColumnProps(kc: KhaosContext, dbName: String, tblName: String, className: String, metaParamsJson: String, tableType: TableType): List[DmTableColumn] = {
    val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)
    val jobProjectId = kc.conf.getInt(SchedulerConstants.PROJECT_ID, 0)
    val jobProjectName = kc.conf.getString(SchedulerConstants.PROJECT_NAME, null)

    //获取物理地址
    var metaRequest = buildMetaRequest(kc)
    var metaParams = new MetaParams(metaParamsJson)
    metaParams.setJobProjectId(jobProjectId);
    metaParams.setJobProjectName(jobProjectName);
    metaRequest.setMetaParams(metaParams)
    if (tableType.equals(TableType.NOSQL)) {
      metaParams.buildNosqlTableParams(runEnv, dbName, tblName)
    } else if (tableType.equals(TableType.RELATION)) {
      metaParams.buildTableParams(runEnv, dbName, tblName)
    } else if (tableType.equals(TableType.FILE)) {
      metaParams.buildFileTableParams(runEnv, dbName, tblName)
    } else {
      metaParams.buildNosqlTableParams(runEnv, dbName, tblName)
    }
//    metaParams.buildNosqlTableParams(runEnv, dbName, tblName)
    val columns = DmMetaUtils.getColumns(className, metaRequest, classOf[List[DmTableColumn]])
    columns
  }

  def main(args: Array[String]): Unit = {

  }

}

